package com.test.mono;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
/**
 * TODO
 *
 * @Description
 * @Author lingmin
 * @Date 2025/4/14 18:22
 **/
@Slf4j
public class MonoPipelineExample {
    public static void main(String[] args) throws InterruptedException {
        // 创建数据处理流水线
        Mono<String> dataPipeline = Mono.fromCallable(() -> {
                    log.info("数据生成线程: " + Thread.currentThread().getName());
                    return "abcdef";
                })
                .subscribeOn(Schedulers.boundedElastic()) // 指定生成数据的线程
                .map(data -> {
                    log.info("第一次转换线程: " + Thread.currentThread().getName());
                    return data.toUpperCase();
                })
                .flatMap(data -> {
                    log.info("异步处理线程: " + Thread.currentThread().getName());
                    return Mono.just(data + " 处理后");
                })
                .publishOn(Schedulers.single()) // 切换后续操作的线程
                .doOnNext(data -> log.info("准备发送数据: " + data))
                .doOnSubscribe(subscription -> log.info("开始订阅"))
                .doOnSuccess(data -> log.info("处理成功"))
                .doOnError(error -> log.error("处理失败: " + error))
                .doFinally(signal -> log.info("最终处理: " + signal));

        // 订阅
        dataPipeline.subscribe(
                data -> log.info("订阅者接收线程: " + Thread.currentThread().getName()
                        + ", 数据: " + data),
                error -> log.error("订阅者错误: " + error)
        );

        // 防止主线程退出太快
        Thread.sleep(1000);
    }
}